package com.atguigu.flink.demo04;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.source.SocketTextStreamFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * @author admin
 * @date 2021/8/10
 */
public class CustomSink {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /**
         * 从网络中读取数据
         * 参数：ip,段口，分隔符，重试次数
         */
        SocketTextStreamFunction socketText = new SocketTextStreamFunction("hadoop102", 9999, "\n", 3);
        DataStreamSource<String> source = env.addSource(socketText);


        // 将数据转换成对对象
        SingleOutputStreamOperator<UserBean> map = source.map(e -> new UserBean(e.split(",")));

        map.addSink(new MySink2());

        map.print();

        env.execute();



    }

    private static class  MySink2 extends RichSinkFunction<UserBean> {



        // 加载驱动类
        private static final String URL = "jdbc:mysql://hadoop102:3306/demo?user=root&password=123321";
        private static final String INSERT_URL="insert into t_user values(?,?,?,?)";

        private  Connection conn;
        private  PreparedStatement statement;

        /**
         * 程序启动是调用
         * @param parameters
         * @throws Exception
         */
        @Override
        public void open(Configuration parameters) throws Exception {
            Class.forName("com.mysql.jdbc.Driver"); // 加载驱动
            conn = DriverManager.getConnection(URL); // 获取数据库连接
            statement = conn.prepareStatement(INSERT_URL);
        }

        @Override
        public void invoke(UserBean value, Context context) throws Exception {


            statement.setInt(1,value.getId());
            statement.setString(2,value.getName());
            statement.setInt(3,value.getAge());
            statement.setString(4,value.getSex());

            statement.execute();

        }

        /**
         * 程序关闭调用
         * @throws Exception
         */
        @Override
        public void close() throws Exception {
            if(statement!=null){
                statement.close();
            }

            if (conn != null) {
                conn.close();
            }
        }
    }


    private static class  MySink implements SinkFunction<UserBean>{

        // 加载驱动类
        private static final String URL = "jdbc:mysql://hadoop102:3306/demo?user=root&password=123321";
        private static final String INSERT_URL="insert into t_user values(?,?,?,?)";


        @Override
        public void invoke(UserBean value, Context context) throws Exception {
            Class.forName("com.mysql.jdbc.Driver"); // 加载驱动
            Connection conn =DriverManager.getConnection(URL); // 获取数据库连接
            PreparedStatement statement = conn.prepareStatement(INSERT_URL);

            statement.setInt(1,value.getId());
            statement.setString(2,value.getName());
            statement.setInt(3,value.getAge());
            statement.setString(4,value.getSex());

            statement.execute();

            statement.close();
            conn.close();
        }

    }
}
